package com.czk.springboot.reactive.reactor.buffer;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.util.List;

public class BufferDemo {

    public static void main(String[] args) {

        Flux<List<Integer>> flux = Flux.range(0, 6)
                // 缓冲区，最大三 每次三个给消费者，类型是Lis
                // [0, 1, 2]
                // [3, 4, 5]
                .buffer(3);

        flux.subscribe(System.out::println);




        flux.subscribe(new BaseSubscriber<>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                // 找发布者请求次数据，具体数据量是由发布者决定，和buffer有关
                request(1);
            }

            @Override
            protected void hookOnNext(List<Integer> value) {
                request(1);
            }

            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                super.hookOnError(throwable);
            }

            @Override
            protected void hookOnCancel() {
                super.hookOnCancel();
            }

            @Override
            protected void hookFinally(SignalType type) {
                super.hookFinally(type);
            }
        });
    }


}
